package com.swak.job.worker;

import java.util.Map;

import com.swak.Constants;
import com.swak.job.Job;
import com.swak.job.JobConfig;
import com.swak.job.enums.Role;
import com.swak.job.queue.QueueService;
import com.swak.leader.ChangeListener;
import com.swak.leader.Leader;
import com.swak.leader.LeaderService;
import com.swak.meters.MetricsFactory;
import com.swak.registry.RegistryService;
import com.swak.registry.URL;
import com.swak.utils.Maps;

public abstract class LeaderAbstractWorker extends AbstractWorker implements ChangeListener {

	private final LeaderService leaderService;
	protected URL leaderUrl;
	protected Leader leader;

	public LeaderAbstractWorker(JobConfig config, Role role, Job job, QueueService queueService,
			RegistryService registryService, LeaderService leaderService, MetricsFactory metricsFactory) {
		super(config, role, job, queueService, registryService, metricsFactory);
		this.leaderService = leaderService;
	}

	protected boolean isLeader() {
		this.leaderSelect();
		return this.leader != null && this.leader.isMe();
	}

	@Override
	protected void doRefresh() {
		this.leaderUrl = URL.valueOf(url).addParameter(Constants.Category_Key, Role.Leader.name());
	}

	@Override
	public void start() {
		super.start();
		this.leaderSelect();
	}

	private void leaderSelect() {
		if (this.leader == null && this.opened) {
			try {
				this.leader = this.leaderService == null ? Leader.Single
						: this.leaderService.leaderSelect(this.leaderUrl.toCategoryPath(), this);
			} catch (Exception e) {
				logger.warn("选主异常", e);
			}
		}
	}

	/**
	 * 更新 Leader 信息
	 */
	@Override
	public void leaderChange(boolean isMe) {
		this.registryService.unregister(this.url);
		Map<String, String> parameters = Maps.newHashMap();
		parameters.put(Leader_key, String.valueOf(isMe));
		this.url = this.initUrl(parameters);
		this.registryService.register(this.url);
	}

	@Override
	public void stop() {
		super.stop();
		try {
			if (this.leader != null) {
				this.leader.leave();
			}
			this.leader = null;
		} catch (Exception e) {
			logger.warn("DispatchWorker 选主异常", e);
		}
	}
}